package com.hu.common.json;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import static org.apache.flink.api.java.typeutils.TypeExtractor.getForClass;

public class JsonDeserializationSchema<T> implements KafkaDeserializationSchema<T> {
    private Class<T> clazz;
    private static ObjectMapper objectMapper = new ObjectMapper();

    public JsonDeserializationSchema(Class<T> clazz) {
        this.clazz = clazz;
    }

    @Override
    public T deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        if (record.value() == null) {
            return null;
        } else {
            T obj = objectMapper.readValue(record.value(), clazz);
            if (obj instanceof TopicPartitionOffset) {
                ((TopicPartitionOffset) obj).setTopic(record.topic());
                ((TopicPartitionOffset) obj).setPartition(record.partition());
                ((TopicPartitionOffset) obj).setOffset(record.offset());
            }
            return obj;
        }
    }

    @Override
    public boolean isEndOfStream(T nextElement) {
        return false;
    }

    @Override
    public TypeInformation<T> getProducedType() {
        return getForClass(clazz);
    }
}
